package com.networkflow_analysis;

import com.networkflow_analysis.bean.ApacheLogEvent;
import com.networkflow_analysis.bean.PageViewCount;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.net.URL;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Map;
import java.util.regex.Pattern;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2021/11/10 14:30
 * <p>
 *
 * 实时热门页面TopN统计（无序） 无法处理迟到数据
 */
public class HotPagesWithList {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //文件源，读取数据，创建dataStream
        /*String filePath = "NetworkFlowAnalysis/src/main/resources/apache.log";
        DataStream<String> fileStream = env.readTextFile(filePath);*/

        /*URL resource = HotPages.class.getResource("/apache.log");
        System.out.println(resource.getPath());
        DataStream<String> fileStream = env.readTextFile(resource.getPath());*/

        //从

        DataStream<String> fileStream = env.socketTextStream("192.168.88.106", 7777);

        /**
         * 数据处理以及进行乱序的watermark的添加
         */
        SingleOutputStreamOperator<ApacheLogEvent> dataStream = fileStream
                .map(line -> {
                    String[] fields = line.split(" ");
                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");
                    Long timestamp = simpleDateFormat.parse(fields[3]).getTime();
                    return new ApacheLogEvent(fields[0], fields[1], timestamp, fields[5], fields[6]);
                })
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<ApacheLogEvent>(Time.seconds(1)) {
                    @Override
                    public long extractTimestamp(ApacheLogEvent apacheLogEvent) {
                        return apacheLogEvent.getTimestamp();
                    }
                });

        dataStream.print("data");
        /**
         * 分组开窗聚合
         */

        //定义一个侧输出流标签
        OutputTag<ApacheLogEvent> lateTag = new OutputTag<ApacheLogEvent>("late") {
        };

        SingleOutputStreamOperator<PageViewCount> windowAggStream = dataStream
                .filter(data -> "GET".equals(data.getMethod()))  //过滤GET请求
                .filter(data -> {
                    String regex = "^((?!\\.(css||js|png||ico)$).)*$";
                    return Pattern.matches(regex, data.getUrl());
                })
//                .keyBy(data -> data.getUrl())
//                .keyBy("url")
                .keyBy(ApacheLogEvent::getUrl)
                .timeWindow(Time.minutes(10), Time.seconds(5))
                .allowedLateness(Time.minutes(1))
                .sideOutputLateData(lateTag)
                .aggregate(new PageCountAgg(), new PageCountResult());


        windowAggStream.print("agg");
        windowAggStream.getSideOutput(lateTag).print("late");

        /**
         * 收集同一窗口count数据，排序输出
         */
        SingleOutputStreamOperator<String> resultStream = windowAggStream
                .keyBy(PageViewCount::getWindowEnd)
                .process(new TopNHotPages(3));

        resultStream.print();

        env.execute("hot pages job");
    }

    /**
     * 自定义预聚合函数
     */
    public static class PageCountAgg implements AggregateFunction<ApacheLogEvent, Long, Long> {

        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(ApacheLogEvent apacheLogEvent, Long aLong) {
            return aLong + 1;
        }

        @Override
        public Long getResult(Long aLong) {
            return aLong;
        }

        @Override
        public Long merge(Long aLong, Long acc1) {
            return aLong + acc1;
        }
    }

    /**
     * 实现自定义的窗口函数
     */
    public static class PageCountResult implements WindowFunction<Long, PageViewCount, String, TimeWindow> {
        @Override
        public void apply(String url, TimeWindow timeWindow, Iterable<Long> iterable, Collector<PageViewCount> collector) throws Exception {
            collector.collect(new PageViewCount(url, timeWindow.getEnd(), iterable.iterator().next()));
        }
    }

    /**
     * 定义自定义处理函数
     */
    public static class TopNHotPages extends KeyedProcessFunction<Long, PageViewCount, String> {

        private Integer topSize;

        public TopNHotPages(Integer topSize) {
            this.topSize = topSize;
        }

        /**
         * 定义一个ListState，保存当前所有的PageViewCount到List
         */
        ListState<PageViewCount> pageViewCountListState;



        @Override
        public void open(Configuration parameters) throws Exception {

            /**
             * ListState
             */
            pageViewCountListState = getRuntimeContext()
                    .getListState(new ListStateDescriptor<PageViewCount>("page-count-list", PageViewCount.class));

        }

        @Override
        public void processElement(PageViewCount pageViewCount, Context context, Collector<String> collector) throws Exception {
            /**
             * ListState
             */
            pageViewCountListState.add(pageViewCount);

            context.timerService().registerEventTimeTimer(pageViewCount.getWindowEnd() + 1);

        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            /**
             * ListState
             */
            ArrayList<PageViewCount> pageViewCounts = Lists.newArrayList(pageViewCountListState.get().iterator());

            pageViewCounts.sort(new Comparator<PageViewCount>() {
                @Override
                public int compare(PageViewCount o1, PageViewCount o2) {
                    if (o1.getCount() > o2.getCount())
                        return -1;
                    else if (o1.getCount() < o2.getCount())
                        return 1;
                    else
                        return 0;
                }
            });

            //格式化为String类型
            StringBuilder resultBuilder = new StringBuilder();

            resultBuilder.append("=================================\n");

            resultBuilder
                    .append("窗口结束时间：")
                    .append(new Timestamp(timestamp - 1))
                    .append("\n");

            //遍历列表，取top n输出
            for (int i = 0; i < Math.min(topSize, pageViewCounts.size()); i++) {

                PageViewCount currentItemViewCount = pageViewCounts.get(i);

                resultBuilder
                        .append("NO").append(i + 1).append(": ")
                        .append(" 页面URL = ").append(currentItemViewCount.getUrl())
                        .append(" 浏览量 = ").append(currentItemViewCount.getCount())
                        .append("\n");
            }
            resultBuilder.append("=============================\n\n");


            //控制输出频率
            Thread.sleep(1000L);

            out.collect(resultBuilder.toString());
        }
    }
}
